package org.fujene.communicate.repli;

import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.Map.Entry;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import org.fujene.communicate.lock.IndexableSwitch;
import org.fujene.communicate.node.Node;
import org.fujene.constants.Constants;
import org.fujene.logging.LogParser;
import org.fujene.structure.Queue;
import org.fujene.structure.ExtendTerm;
import org.fujene.structure.LookupPointer;
import org.msgpack.MessagePackObject;
import org.msgpack.object.ArrayType;
import org.msgpack.object.BooleanType;
import org.msgpack.object.IntegerType;
import org.msgpack.object.RawType;

@SuppressWarnings("unchecked")
public final class ReplicatePeerHandler{
	private static final int TERM = Constants.TERM;
	private static final int CONTENT = Constants.CONTENT;
	private static final int ALL = Constants.ALL;
	private ReplicatePeerSystem mains;
	private int contentNum;
	private int appendixNum;

	public ReplicatePeerHandler(ReplicatePeerSystem peerService, int contentNum, int appendixNum){
		this.mains = peerService;
		this.contentNum = contentNum;
		this.appendixNum = appendixNum;
	}

	/**
	 * Send a node chain.
	 */
	public ArrayType nodeList(){
		LogParser.finest("nodeList - called.");
		int count = 0;
		CopyOnWriteArrayList<Node> currentInfos = mains.getNode();
		MessagePackObject[] output = new MessagePackObject[2 * currentInfos.size()];

		for(Node eachNode: currentInfos){
			output[count++] = RawType.create(eachNode.getAddr());
			output[count++] = IntegerType.create(eachNode.getPort());
		}
		return ArrayType.create(output);
	}

	/**
	 * Registers node to chains.
	 */
	public boolean register(String IPaddr, int port){
		LogParser.finest("register - called.");
		try{
			mains.addNode(Node.create(IPaddr, port));
		}catch(UnknownHostException ignored){
			return false;
		}
		return true;
	}

	/**
	 * @param IPaddr
	 * @param port
	 */
	public void deleteNode(String IPaddr, int port){
		LogParser.finest("deleteNode - called.");
		mains.deleteReplication(IPaddr, port);
	}

	private ArrayList<ExtendTerm> termList = null;
	private ArrayList<Queue> contentList = null;
	private static AtomicInteger counter;
	private static int[] typeTable;
	private static Future<ArrayList<ExtendTerm>>[] termFutures;
	private static Future<ArrayList<Queue>>[] contentFutures;
	private static int[] IDtable;
	private static AtomicInteger[] termCounter;
	private static AtomicInteger[] contentCounter;
	private static ReentrantLock transaction;
	private static final int ARRAY_SIZE = 256;

	static{
		counter = new AtomicInteger(0);
		typeTable = new int[ARRAY_SIZE];
		termFutures = new Future[ARRAY_SIZE];
		contentFutures = new Future[ARRAY_SIZE];
		termCounter = new AtomicInteger[ARRAY_SIZE];
		contentCounter = new AtomicInteger[ARRAY_SIZE];
		IDtable = new int[ARRAY_SIZE];
		transaction = new ReentrantLock();
	}

	/**
	 * Send datas specified by IDs.
	 */
	public ArrayType getData(int ID){
		LogParser.finest("getData - called.");
		boolean isValid = false;
		MessagePackObject[] dummyArray = new MessagePackObject[]{ BooleanType.create(false) };
		// check whether recieved ID's valid
		for(int i = 0; i < ARRAY_SIZE && !isValid; i++)
			if(IDtable[i] == ID) isValid = true;
		int index = ID & 0xFF;

		try{
			if(!isValid){
				// Data not locked
				LogParser.severe("ERROR(getData): Called with invalid ID");
				return ArrayType.create(dummyArray);
			}
			int fieldLength, outputLength;
			MessagePackObject[] output;
			if(typeTable[index] == TERM){
				try{
					if(termList == null) termList = termFutures[index].get();
					if(termCounter[index] == null) termCounter[index] = new AtomicInteger(0);
					// No more data
					if(termCounter[index].intValue() >= termList.size()) return ArrayType.create(dummyArray);
					fieldLength = termList.get(termCounter[index].intValue()).firstIDs.length;
					outputLength = 1 + 3 * fieldLength;
					output = new MessagePackObject[outputLength];
					output[0] = RawType.create(termList.get(termCounter[index].intValue()).term);
					for(int i = 0; i < fieldLength; i++){
						output[1 + 3 * i] = IntegerType
								.create(termList.get(termCounter[index].intValue()).firstIDs[i]);
						// add pointers
						if(termList.get(termCounter[index].intValue()).invIndex[i] != null){
							output[2 + 3 * i] = IntegerType
									.create(termList.get(termCounter[index].intValue()).invIndex[i].block);
							output[3 + 3 * i] = IntegerType
									.create(termList.get(termCounter[index].intValue()).invIndex[i].point);
						}else{
							output[2 + 3 * i] = IntegerType.create(-1);
							output[3 + 3 * i] = IntegerType.create(-1);
						}
					}
				}finally{
					termCounter[index].incrementAndGet();
				}
			}else{
				try{
					if(contentList == null) contentList = contentFutures[index].get();
					if(contentCounter[index] == null) contentCounter[index] = new AtomicInteger(0);
					if(contentList.size() <= contentCounter[index].intValue())
						return ArrayType.create(dummyArray);
					outputLength = 1 + contentList.get(contentCounter[index].intValue()).contents.length
							+ contentList.get(contentCounter[index].intValue()).appendices.length;
					output = new MessagePackObject[outputLength];

					output[0] = IntegerType.create(contentList.get(contentCounter[index].intValue()).ID);
					// Contents
					for(int i = 0; i < contentList.get(contentCounter[index].intValue()).contents.length; i++)
						output[i + 1] = RawType
								.create(contentList.get(contentCounter[index].intValue()).contents[i]);
					// Appendices
					for(int i = 0; i < contentList.get(contentCounter[index].intValue()).appendices.length; i++)
						output[i + 1 + contentList.get(contentCounter[index].intValue()).contents.length] = RawType
								.create(contentList.get(contentCounter[index].intValue()).appendices[i]);
				}finally{
					contentCounter[index].incrementAndGet();
				}
			}
			System.out.println("Output length: " + output.length);
			return ArrayType.create(output);
		}catch(Exception e){
			e.printStackTrace();
		}
		return ArrayType.create(dummyArray);
	}

	/**
	 * 指定した範囲にあるデータサイズを取得し、データ転送の準備をします。
	 */
	public int getDataSize(int type, long hashFrom, long hashTo, String IPaddr, int port, boolean snapshot,
			int snaptype){
		LogParser.finest("getDataSize - called.");
		int ID = -1;
		try{
			transaction.lock();
			int newIndex = counter.getAndIncrement();
			if(newIndex >= ARRAY_SIZE){
				newIndex = 0;
				counter.set(0);
			}
			ID = ((IPaddr.hashCode() + port) << 8) + newIndex;
			// System.out.printf("ID: 0x%08x, newIndex: %d%n", ID, newIndex);
			IDtable[newIndex] = ID;
			typeTable[newIndex] = type;

			// get a snapshot / forced to stop index
			IndexableSwitch.lock();
			if(snaptype == ALL || snaptype == CONTENT)
				contentFutures[newIndex] = mains.getContentSnapshot(hashFrom, hashTo, 0);
			if(snaptype == ALL || snaptype == TERM)
				termFutures[newIndex] = mains.getTermSnapshot(hashFrom, hashTo, 0);
		}finally{
			transaction.unlock();
			try{
				IndexableSwitch.unlock();
			}catch(InterruptedException e){}
		}

		return ID;
	}

	/**
	 * Call to be ended sending data.
	 */
	// FIXME: 手動でロックを外せる or タイムアウト, タイムアウト時間はどうする？
	public boolean endDataSend(int ID){
		LogParser.finest("endDataSend - called.");
		int index = ID & 0xFF;

		if(typeTable[index] == TERM){
			termFutures[index].cancel(true);
			termFutures[index] = null;
			termCounter[index].set(0);
			termList = null;
		}else{
			contentFutures[index].cancel(true);
			contentFutures[index] = null;
			contentCounter[index].set(0);
			contentList = null;
		}
		typeTable[index] = -1;
		IDtable[index] = -1;
		try{
			IndexableSwitch.unlock();
		}catch(InterruptedException e){}
		return true;
	}

	/**
	 * 強制的にデータ転送を中止します。
	 * @param req
	 */
	public void forceStop(int ID){
		LogParser.finest("forceStop - called.");
		int index = ID & 0xFF;

		termFutures[index].cancel(true);
		termFutures[index] = null;
		contentFutures[index].cancel(true);
		contentFutures[index] = null;
		termCounter[index].set(0);
		contentCounter[index].set(0);
		typeTable[index] = -1;

		try{
			IndexableSwitch.unlock();
		}catch(InterruptedException e){}
	}

	public void deleteContent(long ID){
		LogParser.finest("deleteContent - called.");
		mains.invokeDelete(ID);
	}

	public String isLive(){
		LogParser.finest("isLive - called.");
		return "Live";
	}

	/**
	 * Add specified content to the index.
	 */
	public void indexContent(ArrayType content, long ID){
		LogParser.finest("indexContent - called.");
		MessagePackObject[] argobjs = content.asArray();
		String[] contents = new String[contentNum];
		String[] appendices = new String[appendixNum];
		int i;

		for(i = 0; i < contentNum; i++)
			contents[i] = argobjs[i].asString();
		for(; i < contentNum + appendixNum; i++)
			appendices[i - contentNum] = argobjs[i].asString();

		mains.invokeIndexContent(new Queue(contents, appendices, ID));
	}

	/**
	 * Add specified term to the index.
	 */
	public static int terms = 0;

	public void indexTerm(String term, int termfreq, long ID, int docType){
		LogParser.finest("IndexTerm - called.");
		mains.invokeIndexTerm(term, termfreq, ID, docType);
	}

	/**
	 * Search term from own index.
	 */
	public byte[] searchTerm(String query, int docType){
		LogParser.finest("searchTerm - called.");
		ExtendTerm result = mains.invokeSearch(query, docType);

		if(result == null) return null;
		return result.createByteArray();
	}

	/**
	 * Send content from specified ID.
	 */
	public ArrayType getContent(long ID, int docType){
		LogParser.finest("getContent - called.");
		int i;
		Queue content = mains.lookupContent(ID, docType);
		MessagePackObject[] retobjs = new MessagePackObject[content.contents.length
				+ (content.appendices == null ? 0 : content.appendices.length)];

		for(i = 0; i < content.contents.length; i++){
			retobjs[i] = RawType.create(content.contents[i]);
		}
		for(i = 0; i < content.appendices.length; i++){
			retobjs[i + content.contents.length] = RawType.create(content.appendices[i]);
		}

		return ArrayType.create(retobjs);
	}

	/**
	 * 本文のIDポインタを取得します。
	 */
	public ArrayType getList(int ptrBlock, int ptrPoint, long IDfrom, int length){
		LogParser.finest("getList - called.");
		LookupPointer ptr = new LookupPointer(ptrBlock, ptrPoint, IDfrom);
		LinkedHashMap<Long, Integer> list = mains.getContentsByHash(ptr, length);

		MessagePackObject[] retval = new MessagePackObject[2 * list.size() + 3];
		retval[0] = IntegerType.create(ptr.block);
		retval[1] = IntegerType.create(ptr.point);
		retval[2] = IntegerType.create(ptr.firstID);
		int i = 0;
		for(Entry<Long, Integer> entry: list.entrySet()){
			retval[2 * i + 3] = IntegerType.create(entry.getKey().longValue());
			retval[2 * i + 4] = IntegerType.create(entry.getValue().intValue());
			i++;
		}

		return ArrayType.create(retval);
	}

	/**
	 * ノード参加に必要な情報を各種送信します。 (BlockNum, BlockSize, Replication, IndexMode)
	 */
	public ArrayType getInfo(){
		LogParser.finest("getInfo - called.");
		return mains.putInfo();
	}

	public int getIndexQueue(){
		return mains.getQueueSize(0);
	}

	public int getSearchQueue(){
		return mains.getQueueSize(1);
	}

	public void stopIndexing(){
		IndexableSwitch.lock();
	}

	public void resumeIndexing(){
		try{
			IndexableSwitch.unlock();
		}catch(InterruptedException e){}
	}

	public void revertLatest(){
		mains.revert(-1);
	}

	public void backupLatest(){
		mains.backup();
	}

	public void snapshot(){
		mains.takeSnapshot();
	}

	public void synchronizeClock(){
		mains.synchronizeClock();
	}

	public void suspend(String IPaddr, int port, long suspendTime){
		mains.suspend(IPaddr, port, suspendTime);
	}
}
